package com.xujianlong.app.func;

import com.alibaba.fastjson.JSONObject;
import com.xujianlong.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import com.xujianlong.utils.DimUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Connection connection;
    @Override
    public void open(Configuration parameters) throws Exception {
        //TODO 1.获取Phoenix连接
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //value:{"database":"gmall-210726-flink","before":{},"after":{"tm_name":"shanghai","id":17},"type":"insert","tableName":"base_trademark","sinkTable":"dim_base_trademark"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //1.准备SQL语句:upsert into db.tn(id,name,sex) values('1001','zhangsan','male')
        String sinkTable = value.getString("sinkTable");
        JSONObject after = value.getJSONObject("after");
        String sql = genSql(sinkTable,after);
        //TODO 2.通过连接创建PrepareStatment,并注入sql
        PreparedStatement preparedStatement = connection.prepareStatement(sql);


        //TODO redis清除数据
        if("update".equals(value.getString("type"))){
            DimUtil.delDimInfo(sinkTable.toUpperCase(),after.getString("id"));
        }

        //TODO 3.PrepareStatment执行代码
        preparedStatement.execute();

        //TODO 4.Phoenix默认自动提交为false
        connection.commit();

        //TODO 5.释放资源
        preparedStatement.close();
    }

    private String genSql(String sinkTable,JSONObject after){

        Set<String> columns=after.keySet();
        Collection<Object> values = after.values();

        return "upsert into "+GmallConfig.HBASE_SCHEMA+"."+sinkTable+
                "("+ StringUtils.join(columns,",")+")"+
                "values('"+StringUtils.join(values,"','")+"')";

    }
}
